package com.feidee.fd.sml.algorithm.component.feature

import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.feature.IndexToString
import org.apache.spark.sql.DataFrame

/**
  * @Author tangjinyuan
  * @Date 2019/03/21 16:33
  * @Description IndexToStringEncoder 索引转字符串 特征组件
  * @Reviewer dongguosheng
  */
case class IndexToStringEncoderParam(
                                      override val input_pt: String,
                                      override val output_pt: String,
                                      override val hive_table: String,
                                      override val flow_time: String,
                                      override val inputCol: String,
                                      override val outputCol: String,
                                      override val preserveCols: String,
                                      override val modelPath: String,
                                      // 设置 lables 数组，推荐但非必填
                                      labels: Array[String]
                                    ) extends FeatureParam {

  def this() = this(null, null, null, null, "input", "features", null, null, Array.empty[String])

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("labels" -> labels)
    map
  }
}


class IndexToStringEncoder extends AbstractFeatureEncoder[IndexToStringEncoderParam] {

  override def setUp(param: IndexToStringEncoderParam, data: DataFrame): Array[PipelineStage] = {

    val indexToString = new IndexToString()
      .setLabels(param.labels)
      .setInputCol(param.inputCol)
      .setOutputCol(param.outputCol)

    Array(indexToString)
  }

}


object IndexToStringEncoder {

  def apply(paramStr: String): Unit = {
    new IndexToStringEncoder()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    IndexToStringEncoder(args(0))
  }
}